package org.apache.rocketmq.proxy.service.route;

import com.google.common.collect.Lists;
import com.google.common.net.HostAndPort;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.client.impl.mqclient.MQClientAPIFactory;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.proxy.common.Address;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
import org.apache.rocketmq.remoting.Configuration;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;

import java.util.HashMap;
import java.util.List;

/**
 * @author: jiaxiangen
 * @Date: 2025/6/23 17:45
 * @Version 1.0
 * @Description: 本地主题路由服务
 */
public class LocalTopicRouteService extends TopicRouteService {

    private final BrokerController brokerController;

    private final List<BrokerData> brokerDataList;

    private final int grpcPort;

    public LocalTopicRouteService(BrokerController brokerController, MQClientAPIFactory mqClientAPIFactory){
        super(mqClientAPIFactory);
        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
        this.brokerController = brokerController;
        HashMap<Long, String> brokerAddrs = new HashMap<>();
        brokerAddrs.put(MixAll.MASTER_ID, brokerController.getBrokerAddr());
        BrokerData brokerData = new BrokerData(brokerController.getBrokerConfig().getBrokerClusterName(), brokerController.getBrokerConfig().getBrokerName(), brokerAddrs);
        this.brokerDataList = Lists.newArrayList(brokerData);
        grpcPort = proxyConfig.getGrpcServerPort() ;
    }


    @Override
    public MessageQueueView getCurrentMessageQueueView(ProxyContext ctx, String topicName) throws Exception {
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().getTopicConfigTable().get(topicName);
        return new MessageQueueView(topicName, toTopicRouteData(topicConfig), null);
    }

    private TopicRouteData toTopicRouteData(TopicConfig topicConfig) {
        TopicRouteData topicRouteData = new TopicRouteData();
        topicRouteData.setBrokerDatas(brokerDataList);
        QueueData queueData = new QueueData();
        queueData.setBrokerName(brokerController.getBrokerConfig().getBrokerName());
        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
        queueData.setPerm(topicConfig.getPerm());
        queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());
        topicRouteData.setQueueDatas(Lists.newArrayList(queueData));
        return topicRouteData;
    }

    @Override
    public ProxyTopicRouteData getTopicRouteForProxy(ProxyContext ctx, List<Address> requestHostAndPortList,
                                                     String topicName) throws Exception {
        //从Proxy中获取本地缓存的指定主题下的路由信息视图
        MessageQueueView messageQueueView = getAllMessageQueueView(ctx, topicName);
        //从路由信息视图中获取路由信息
        TopicRouteData topicRouteData = messageQueueView.getTopicRouteData();
        //创建一个代理主题路由数据对象
        ProxyTopicRouteData proxyTopicRouteData = new ProxyTopicRouteData();
        //把路由信息中的队列数据设置到代理主题路由数据对象中
        proxyTopicRouteData.setQueueDatas(topicRouteData.getQueueDatas());
        //在这里得到路由信息中的Broker数据，注意，这里得到的是BrokerData中封装着每一个Broker主从集群，或者主从组的信息
        for (BrokerData brokerData : topicRouteData.getBrokerDatas()) {
            //创建一个代理主题路由数据中的代理Broker数据对象，这个对象将会封装着每一个Broker主从组的信息
            ProxyTopicRouteData.ProxyBrokerData proxyBrokerData = new ProxyTopicRouteData.ProxyBrokerData();
            //设置代理Broker数据对象的集群名称
            proxyBrokerData.setCluster(brokerData.getCluster());
            //设置代理Broker数据对象的Broker名称
            proxyBrokerData.setBrokerName(brokerData.getBrokerName());
            //在这里遍历每一个Broker主从组中，主节点和从节点的Id
            for (Long brokerId : brokerData.getBrokerAddrs().keySet()) {
                //根据主从节点Id，得到每一个节点的网络地址
                String brokerAddr = brokerData.getBrokerAddrs().get(brokerId);
                //在这里得到了每一个Broker网络地址HostAndPort对象
                //请大家注意一点，因为目前是在local模式下部署Proxy和Broker的，这两个模块部署在同一个服务器上了
                //所以这里得到的Broker的网络地址，其实就是Proxy的网络地址，当然，它们的端口号肯定是不同的，但是IP地址是一样的
                HostAndPort brokerHostAndPort = HostAndPort.fromString(brokerAddr);
                //这里就是把Proxy服务器监听的端口号和Broker的网络地址拼接到一起，实际上就是得到了Proxy服务器的网络地址
                //路由信息被返回给客户端的时候，客户端就可以根据这个网络地址，来访问Proxy服务器了，这个原因我已经为大家解释过了
                //在5.0之前的版本，客户端是和Broker直接通信的，但是在5.0之后的版本，客户端是和Proxy服务器通信的，所以Proxy返回的路由信息中的Broker的地址
                //都要替换成Proxy的地址，客户端发送消息和消费消息仍然是访问Proxy，Proxy再转发给Broker，看到这里大家应该也可以意识到，这么一来的话
                //负载均衡，真正的路由操作其实是在Proxy服务器上进行的，Broker只是一个存储消息的地方，真正的路由操作是在Proxy服务器上进行的
                HostAndPort grpcHostAndPort = HostAndPort.fromParts(brokerHostAndPort.getHost(), grpcPort);
                //把代理Broker数据对象中的Broker网络地址设置为Proxy服务器的网络地址
                proxyBrokerData.getBrokerAddrs().put(brokerId, Lists.newArrayList(new Address(Address.AddressScheme.IPv4, grpcHostAndPort)));
            }//把代理Broker数据对象添加到代理主题路由数据对象中
            proxyTopicRouteData.getBrokerDatas().add(proxyBrokerData);
        }
        return proxyTopicRouteData;
    }


    @Override
    public String getBrokerAddr(ProxyContext ctx, String brokerName) throws Exception {
        return this.brokerController.getBrokerAddr();
    }
}
